package lab82;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.io.Text;

/**
 * @author wangcheng
 *         需求：山东省将要在各大城市建造飞机场，政府提供2份数据，1份是所有城市，1份是已经建飞机场的城市，通过MapReduce计算目前还没有飞机场的城市
 *         思路：写2个Mapper，分别来读取这2份数据，让这2个Mapper处理完的结果连接，进入同一个Shuffle之后，再在Reduce端进行最终的过滤输出
 *         allCity.txt 青岛 淄博 烟台 德州 济南 泰安 滨州 莱芜 日照 东营 菏泽 聊城 济宁 临沂 枣庄 威海 潍坊
 *         someCity.txt 济南 济南机场 青岛 青岛机场 潍坊 潍坊机场 临沂 临沂机场 济宁 济宁机场 威海 威海机场 东营 东营机场
 */

public class CityMapJoinDemo {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        // 判断输入参数是否为3个，2个是Mapper的输入，第3个是Reduce输出
        if (args.length != 3 || args == null) {
            System.err.print("Please input Full Path!");
            System.exit(1);
        }

        // 创建作业对象，运行的主类
        Job job = Job.getInstance(new Configuration(), "CityMapJoinDemo");
        job.setJarByClass(CityMapJoinDemo.class);

        // 添加所有城市文件路径及其对应的AllCityMapper处理
        MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, AllCityMapper.class);
        // 添加已建立飞机场城市的文件路径及其对应的SomeCityMapper处理
        MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class, SomeCityMapper.class);
        FileOutputFormat.setOutputPath(job, new Path(args[2]));
        // 设置处理2个Mapper输出结果的Reducer处理CityReduce
        job.setReducerClass(CityReduce.class);

        // 设置job输出的key类型为Text
        job.setOutputKeyClass(Text.class);
        // 设置job输出的value类型为Text
        job.setOutputValueClass(Text.class);

        // 提交运行job
        int num = job.waitForCompletion(true) ? 0 : 1;
        // 根据job执行返回的结果退出程序
        System.exit(num);
    }
}
